package com.lagou.client;

import com.lagou.constant.Constant;
import com.lagou.rpc.*;
import com.lagou.service.UserService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer {
    private static Logger logger = LoggerFactory.getLogger(RpcConsumer.class);
    //创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static Map<String, Bootstrap> serviceRpcMap = new HashMap<>();

    public static CuratorFramework curatorFramework;

    public static List<String> serviceUrl = new ArrayList<>();

    public static Map<String, UserClientHandler> serviceCacheMap = new HashMap<>();

    public static Map<UserClientHandler, String> handlerServiceNameMap = new HashMap<>();

    public static Map<String, EventLoopGroup> bootstrapMap = new HashMap<>();

    public static int callCount = 0;

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    public Object createProxy(final Class<?> serviceClass, final String providerName) {
        //借助JDK动态代理生成代理对象
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                //（1）调用初始化netty客户端的方法

                if (serviceCacheMap.size() == 0) {
                    initZk();
                }

                // 设置参数
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setClassName(UserService.class.getName());
                rpcRequest.setMethodName("sayHello");
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                String methodParam = "hello";
                rpcRequest.setParameters(args);
                Class[] parameterTypes = new Class[]{};
                if (null != args || args.length != 0) {
                    parameterTypes = new Class[args.length];
                    for (int i = 0; i < args.length; i++) {
                        parameterTypes[i] = args[i].getClass();

                    }
                }
                rpcRequest.setParameterTypes(parameterTypes);

                UserClientHandler userClientHandler = getHandlerByVisitTime();
                if(null == userClientHandler){
                    logger.error("无可用服务!");
//                    throw new RuntimeException("无可用服务");
                    return null;
                }else{
                    userClientHandler.setPara(rpcRequest);
                }

                // 去服务端请求数据
                //TODO 为啥提交到线程池中，要不要每次都new一个userClientHandler？
                return executor.submit(userClientHandler).get();
            }
        });


    }
//轮训调用不同服务
    private UserClientHandler getHandler() {
        if(serviceCacheMap.size() == 0)
            return null;
        List<UserClientHandler> handlers = new ArrayList<>(serviceCacheMap.values());
        UserClientHandler handler = handlers.get(callCount++ % handlers.size());
        String serviceName = handlerServiceNameMap.get(handler);
        logger.info("调用服务{}开始", serviceName);
        return handler;
    }

    //轮训调用不同服务
    private UserClientHandler getHandlerByVisitTime() {
        String path = Constant.SERVIE_PATH_ZK + "/" + Constant.USER_SERVICE_NAME + "/" + Constant.SERVICE_LAST_VISITED;
        try {
            Optional<List<String>> optionalStrings = Optional.of(curatorFramework.getChildren().forPath(path));
            Map<String, String> nodeVisitMap = new HashMap<>();
            List<String> visitTimes = new ArrayList<>();
            for (String child : optionalStrings.orElse(new ArrayList<String>())) {
                String childPath = path + '/' + child;
                byte[] b = curatorFramework.getData().forPath(childPath);
                String lastVisit = new String(b, "utf-8");

                String[] strings = lastVisit.split("=");
                nodeVisitMap.put(strings[1], strings[0]);
                visitTimes.add(strings[1]);
                logger.info("获取到服务的最后调用时间：服务名{}时间{}", strings[0], strings[1]);
            }
            Collections.sort(visitTimes);
            if(visitTimes.size() != 0){
                String visitTime = visitTimes.get(visitTimes.size() -1);
                DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Date visitDate = df.parse(visitTime);
                long visitDateLong = visitDate.getTime();
                if(System.currentTimeMillis() - visitDateLong <= 3000000){

                    String serviceNodeName = nodeVisitMap.get(visitTime);
                    logger.info("调用服务{}开始", serviceNodeName);
                    return serviceCacheMap.get(serviceNodeName);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return getHandler();
    }

    public static void initZk() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework curator = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
        curatorFramework = curator;
        curatorFramework.start();
        curatorFramework = curator;

        String path = Constant.SERVIE_PATH_ZK + "/" + Constant.USER_SERVICE_NAME + "/" + Constant.SERVICE_NODE;
        try {
            Optional<List<String>> optionalStrings = Optional.of(curatorFramework.getChildren().usingWatcher(new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {

                }
            }).forPath(path));
            PathChildrenCache nodeCache = new PathChildrenCache(curatorFramework, path,true);
            //调用start方法开始监听
            nodeCache.start();
            //添加NodeCacheListener监听器
            nodeCache.getListenable().addListener((client, event) -> {
                logger.info("节点数据变化，事件类型", path);
                try {
                    Optional<List<String>> optionalStrings1 = Optional.of(curatorFramework.getChildren().forPath(path));
                    for (String child : optionalStrings1.orElse(new ArrayList<>())) {
                        if(!serviceCacheMap.containsKey(child)){
                            logger.info("有新服务{}上线,启动连接", child);
                            initRpcClient(child, path);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            for (String child : optionalStrings.orElse(new ArrayList<String>())) {
                initRpcClient(child, path);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void initRpcClient(String child, String path) throws Exception {
        String childPath = path + '/' + child;
        byte[] b = curatorFramework.getData().usingWatcher(new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if(watchedEvent.getType() == Event.EventType.NodeDeleted){
                    EventLoopGroup group = bootstrapMap.get(child);
                    if(null != group)
                        group.shutdownGracefully();
                    bootstrapMap.remove(child);
                    UserClientHandler handler = serviceCacheMap.get(child);
                    serviceCacheMap.remove(child);
                    handlerServiceNameMap.remove(handler);
                    logger.info("服务{}已下线，从缓存列表中移除",child);
                }
            }
        }).forPath(childPath);
        String serviceUrl = new String(b, "utf-8");
        if (null != serviceUrl) {
            logger.info("服务{}已上线，初始化连接", serviceUrl);
            UserClientHandler userClientHandler = new UserClientHandler();
            initClient(serviceUrl, userClientHandler, child);
            //加入缓存中
            RpcConsumer.serviceUrl.add(serviceUrl);
            serviceCacheMap.put(child, userClientHandler);//childPath
            handlerServiceNameMap.put(userClientHandler, child);
        }
    }


    //2.初始化netty客户端
    public static Bootstrap initClient(String serviceUrl, UserClientHandler userClientHandler, String node) throws InterruptedException {
        String[] serviceConfig = serviceUrl.split(":");
        String host = serviceConfig[0];
        int port = Integer.parseInt(serviceConfig[1]);

        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandler);
                    }
                });

        bootstrap.connect(host, port).sync();
        bootstrapMap.put(node, group);
        return bootstrap;
    }
}
